package com.wlyuan.index.domain.index.service;

import cn.hutool.core.date.StopWatch;
import com.wlyuan.core.domain.valueobj.OrderId;
import com.wlyuan.index.domain.index.factory.OrderIndexFactory;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.apache.commons.collections4.CollectionUtils;
import org.redisson.api.RSet;
import org.redisson.api.RedissonClient;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
@RequiredArgsConstructor
public class OrderIndexRenewService {
    private final OrderIndexFactory indexFactory;
    private final RedissonClient redissonClient;
    private static final String CACHE_NAME = "PC::CACHE::INDEX::ORDER::TASK";
    private static final ExecutorService EXECUTOR = new ThreadPoolExecutor(1,
            1, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(),
            new CustomizableThreadFactory("order-index-renew-pool"));
    private RSet<OrderId> tasks;
    private int sleepMillis;

    @PostConstruct
    public void initialize() {
        this.tasks = redissonClient.getSet(CACHE_NAME);
        EXECUTOR.execute(() -> {
            //noinspection InfiniteLoopStatement
            while (true) {
                try {
                    this.loop();
                } catch (Exception e) {
                    logger.error("刷新订单索引异常", e);
                }

            }
        });
    }

    public void add(OrderId orderId) {
        this.tasks.addAsync(orderId);
    }

    public void addAll(Collection<OrderId> orderIds) {
        this.tasks.addAllAsync(orderIds);
    }

    protected Collection<OrderId> take() {
        // 随机取50条
        var orderIds = this.tasks.random(50);
        if (CollectionUtils.isEmpty(orderIds)) {
            return Collections.emptySet();
        }
        // 移除取出来的任务
        this.tasks.removeAll(orderIds);
        return orderIds;
    }

    private void sleep() {
        try {
            if (this.sleepMillis < 500) {
                this.sleepMillis++;
            }
            Thread.sleep(this.sleepMillis);
        } catch (InterruptedException e) {
            logger.error("Thread sleep error", e);
        }
    }

    private void renew(Collection<OrderId> orderIds) {
        if (CollectionUtils.isEmpty(orderIds)) {
            this.sleep();
            return;
        }
        StopWatch watch = new StopWatch();
        if (logger.isDebugEnabled()) {
            watch.start("刷新订单索引");
        }
        orderIds.forEach(indexFactory::create);
        if (logger.isDebugEnabled()) {
            watch.stop();
        }
        if (logger.isDebugEnabled()) {
            logger.debug("刷新订单索引耗时: {} {}", orderIds.size(), watch.prettyPrint(TimeUnit.MILLISECONDS));
        }
    }

    private void loop() {
        var orderIds = this.take();
        try {
            this.renew(orderIds);
        } catch (Exception e) {
            logger.error("刷新订单索引异常: {}", orderIds, e);
        }
    }
}
